package com.study.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @Description: 批处理wordcount统计
 * @Author: LiuQun
 * @Date: 2022/7/24 15:04
 */
public class BatchWordCount {
    /**
     * 下面这种方式使用的是DataSet API进行批处理操作，在flink 1.12之后就开始弃用这种方式
     */
    public static void main(String[] args) throws Exception {

        //1.创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. 从文件读取数据  按行读取(存储的元素就是每行的文本)
        DataSource<String> lineDS = env.readTextFile("input/words.txt");

        // 3.将每行数据进行分词，转换成二元组类型
        FlatMapOperator<String, Tuple2<String, Long>> wordAnd1Tuple = lineDS.flatMap(
                //line是文本中每行数据
                (String line, Collector<Tuple2<String, Long>> out) -> {
                    //及那个一行文本进行分词
                    String[] wordArr = line.split(" ");
                    //将每个单词转换成二元组输出
                    for (String word : wordArr) {
                        out.collect(Tuple2.of(word, 1L));
                    }
                }
        ).returns(Types.TUPLE(Types.STRING, Types.LONG)); //因为泛型擦除，因此需要在返回时指定数据类型

        // 4.按照word进行分组
        UnsortedGrouping<Tuple2<String, Long>> wordAnd1Group = wordAnd1Tuple.groupBy(0);//元组中每一个元素都有一个索引，按第一个索引分组即按照word分组

        // 5.分组内进行聚合统计
        AggregateOperator<Tuple2<String, Long>> sum = wordAnd1Group.sum(1); //将元组中的第二个元素进行统计

        // 6.打印结果
        sum.print();

    }
}
